package reader.file;

import ch.qos.logback.classic.Level;
import ch.qos.logback.classic.LoggerContext;
import core.Job;
import reader.file.monitor.FileListener;
import reader.file.monitor.FileMonitor;
import org.slf4j.LoggerFactory;
import writer.es.DataBeanEsConsumer;
import writer.file.FileCreateTest;

import java.io.File;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;

public class FileJob implements Job {
    private FileProperties fileProperties;
    private ExecutorService pool;
    private FileMonitor fileMonitor;
    private BlockingQueue<Map<String, File>> queue;

    private static CountDownLatch latch = new CountDownLatch(1024);

    static {
        LoggerContext loggerContext = (LoggerContext) LoggerFactory.getILoggerFactory();
        List<ch.qos.logback.classic.Logger> loggerList = loggerContext.getLoggerList();
        loggerList.forEach(logger -> {
            logger.setLevel(Level.ERROR);
        });
    }

    public void init(FileProperties fileProperties) {
        queue = FileQueue.getQueue();
        String url = fileProperties.getUrlPath(); //"D://file"
        File f = new File(url);
        // 根据已存在的文件总数全量处理文件
        final List<File> filePathsList = new ArrayList<File>();
        File[] filePaths = f.listFiles();

        for (File s : filePaths) {
            if (s.getAbsolutePath().indexOf(".txt")>-1) {
                filePathsList.add(s);
                Map<String, File> fileMap = new HashMap<String, File>();
                fileMap.put(s.getName(), s);
                queue.add(fileMap);
            }
        }

        // 创建一个文件观察器用于处理文件的格式
        // 设置文件变化监听器
        fileMonitor = new FileMonitor(1000);
        fileMonitor.monitor(url, new FileListener());
        pool = FileProcessorPool.getPool(fileProperties);
    }

    public void start() throws Exception  {
        fileMonitor.start();
        //增量处理文件
        pool.submit(new FileReader(queue, latch));
        pool.submit(new FileReader(queue, latch));
        pool.submit(new FileReader(queue, latch));
        pool.submit(new FileReader(queue, latch));
    }

    public void stop() throws Exception  {
        fileMonitor.stop();
        pool.shutdown();
    }

    public void destroy() {

    }

    public static void main(String[] args) {
        Thread t = new Thread(new FileCreateTest.MyCallableCreate());
        t.start();


        FileProperties fileProperties = new FileProperties("D://file", 4);
        FileJob fileJob = new FileJob();
        fileJob.init(fileProperties);
        try {
            fileJob.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
        Thread dataBeanConsumerThread = new Thread(new DataBeanEsConsumer());
        dataBeanConsumerThread.start();

    }
}
